package cn.liuwenlin.mq;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.ScheduledMessage;

import javax.jms.*;
import java.util.concurrent.TimeUnit;

public class QueueSender {

    public static void main(String[] args) {
        //1.建立连接工厂
//        ActiveMQConnectionFactory activeMQConnectionFactory
//                = new ActiveMQConnectionFactory(
//                        ActiveMQConnectionFactory.DEFAULT_USER,
//                        ActiveMQConnectionFactory.DEFAULT_PASSWORD,
//                "tcp://master.tony.com:61616");
        //使用nio配置的连接传输协议,提高数据传输效率
        ActiveMQConnectionFactory activeMQConnectionFactory
                = new ActiveMQConnectionFactory(
                ActiveMQConnectionFactory.DEFAULT_USER,
                ActiveMQConnectionFactory.DEFAULT_PASSWORD,
                "nio://hadoop.master.com:61617");
        Connection connection = null;
        try {
            //2.获取一个向Active MQ的连接
            connection = activeMQConnectionFactory.createConnection();
            connection.start();
            //3.获取session（会话可以设置是否开启事务，如果开启，则需要在发送消息后做提交操作，否则消息不会投递）
//            Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

            //4.找目的地，获取destination，消费端，也会从这个目的地取消息
            Queue mq = session.createQueue("activemq");

            //5.1消息生产者
            MessageProducer producer = session.createProducer(mq);

            //5.2创建消息
            TextMessage msg = null;
            BytesMessage bytesMessage = null;
            for(int i = 0; i < 100; i++){
                msg = session.createTextMessage("Hi activemq" + i);

                //设置消息的优先级
                if(i % 10 == 0){
                    bytesMessage = session.createBytesMessage();
                    bytesMessage.writeBytes("你好 mq".getBytes());
                    bytesMessage.setDoubleProperty("ttl", 200.0 + i * 10);
                    //设置持久化，优先级和ttl时间
                    producer.send(bytesMessage, DeliveryMode.NON_PERSISTENT, 9, 1000);
//                    producer.setTimeToLive(1000);
//                    producer.send(bytesMessage);
                    System.out.println(bytesMessage + " has been send...");
                    continue;
                }

                //延迟消息投递(支持 cron 表达式)
                if(i % 5 == 0){
                    //延迟投递时间
                    msg.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, 1000 * 10);
                    //重复投递的间隔时间
                    msg.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_PERIOD, 1000 * 2);
                    //手动设置的重复投递次数
                    msg.setIntProperty(ScheduledMessage.AMQ_SCHEDULED_REPEAT, 9);
                    msg.setIntProperty("type", 1);
                    producer.send(msg);
                    continue;
                }

                //5.3向目的地写入消息
                producer.send(msg);
                System.out.println(msg + " has been send...");

//                if(i % 4 == 0){
//                    session.rollback();
//                } else {
//                    session.commit();
//                }

                TimeUnit.MILLISECONDS.sleep(10);
            }


        } catch (JMSException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {

            //6.关闭连接
            if(connection != null){
                try {
                    connection.close();
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }

            System.out.println("exit...");
        }

    }

}
